package com.boot.stream.codec;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Deserializer;

import java.util.Map;

public class JsonDeserializer<T> implements Deserializer<T> {
    private ObjectMapper objectMapper = new ObjectMapper();

    private Class<T> cls;

    private String model;

//    public JsonDeserializer() {
//    }

    public JsonDeserializer(String model) {
        this.model = model;
    }

    @Override
    public void configure(Map<String, ?> props, boolean isKey) {
        this.cls = (Class<T>) props.get(this.model);
    }

    @Override
    public T deserialize(String topic, byte[] bytes) {
        if (bytes == null) {
            return null;
        }
        T data;
        try {
            data = this.objectMapper.readValue(bytes, this.cls);
        } catch (Exception e) {
            throw new SerializationException(e);
        }
        return data;
    }

    @Override
    public void close() {

    }
}
